package com.atguigu.flink.cep;


import com.atguigu.flink.pojo.LoginEvent;
import com.atguigu.flink.pojo.LoginEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2023/3/4
 *
 *  同一用户（可以是不同IP）在2秒之内连续两次登录，都失败，就认为存在恶意登录的风险，输出相关的信息进行报警提示
 */
public class Demo12_LoginExec
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<LoginEvent> watermarkStrategy = WatermarkStrategy
            .<LoginEvent>forBoundedOutOfOrderness(Duration.ofMinutes(1))
            .withTimestampAssigner((e, r) -> e.getEventTime());

        env.setParallelism(1);

        //读取数据
        KeyedStream<LoginEvent, Long> ds = env
            .readTextFile("data/LoginLog.csv")
            .map(new MapFunction<String, LoginEvent>()
            {
                @Override
                public LoginEvent map(String value) throws Exception {
                    String[] words = value.split(",");
                    return new LoginEvent(
                        Long.valueOf(words[0]),
                        words[1],
                        words[2],
                        Long.valueOf(words[3]) * 1000
                    );
                }
            })
            .assignTimestampsAndWatermarks(watermarkStrategy)
            //把同一用户的数据分到一起
            .keyBy(LoginEvent::getUserId);



        //定义Pattern 2秒之内连续两次登录，都失败
        Pattern<LoginEvent, LoginEvent> pattern = Pattern.<LoginEvent>begin("fail")
            .where(new SimpleCondition<LoginEvent>()
            {
                //定义匹配规则
                @Override
                public boolean filter(LoginEvent value) throws Exception {
                    return "fail".equals(value.getEventType());
                }
            })
            .within(Time.seconds(2))
            .times(2)
            //连续两次是 next(紧挨着)关系
            .consecutive()
            ;

        OutputTag<LoginEvent> late = new OutputTag<>("late", TypeInformation.of(LoginEvent.class));

        PatternStream<LoginEvent> patternStream = CEP.pattern(ds, pattern)
                                                      .sideOutputLateData(late);


        SingleOutputStreamOperator<String> ds2 = patternStream.select(new PatternSelectFunction<LoginEvent, String>()
        {

            @Override
            public String select(Map<String, List<LoginEvent>> map) throws Exception {
                List<LoginEvent> events = map.get("fail");
                Long userId =events.get(0).getUserId();
                List<Long> ts = events.stream().map(e -> e.getEventTime()).collect(Collectors.toList());

                return userId +"涉嫌恶意登录，时间是:" + ts;
            }
        });
        ds2
                     .print().setParallelism(1);
        ds2.getSideOutput(late).printToErr();


       env.execute();

    }
}
